package drds.binlog.parse.inbound.mysql;

import drds.binlog.common.position.EntryPosition;
import drds.binlog.filter.EventFilter;
import drds.binlog.filter.aviater.AviaterRegexFilter;
import drds.binlog.parse.EventParser;
import drds.binlog.parse.exception.BinlogParseException;
import drds.binlog.parse.inbound.AbstractEventParser;
import drds.binlog.parse.inbound.BinlogParser;
import drds.binlog.parse.inbound.IMultiStageCoprocessor;
import drds.binlog.parse.inbound.mysql.dbsync.LogEventConvert;
import drds.binlog.parse.inbound.mysql.tsdb.DefaultTableMetaTSDBFactory;
import drds.binlog.parse.inbound.mysql.tsdb.TableMetaTSDB;
import drds.binlog.parse.inbound.mysql.tsdb.TableMetaTSDBFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.concurrent.atomic.AtomicLong;

public abstract class AbstractMysqlEventParser extends AbstractEventParser
{

    protected static final long BINLOG_START_OFFEST = 4L;
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    private final AtomicLong eventsPublishBlockingTime = new AtomicLong(0L);
    protected TableMetaTSDBFactory tableMetaTSDBFactory = new DefaultTableMetaTSDBFactory();
    protected boolean enableTsdb = false;
    protected String tsdbSpringXml;
    protected TableMetaTSDB tableMetaTSDB;
    // 编码信息
    protected byte connectionCharsetNumber = (byte) 33;
    protected Charset connectionCharset = Charset.forName("UTF-8");
    protected boolean filterQueryDcl = false;
    protected boolean filterQueryDml = false;
    protected boolean filterQueryDdl = false;
    protected boolean filterRows = false;
    protected boolean filterTableError = false;


    protected BinlogParser buildParser()
    {
        LogEventConvert logEventConvert = new LogEventConvert();
        if (eventFilter != null && eventFilter instanceof AviaterRegexFilter)
        {
            logEventConvert.setNameFilter((AviaterRegexFilter) eventFilter);
        }

        if (eventBlackFilter != null && eventBlackFilter instanceof AviaterRegexFilter)
        {
            logEventConvert.setNameBlackFilter((AviaterRegexFilter) eventBlackFilter);
        }

        logEventConvert.setCharset(connectionCharset);
        logEventConvert.setFilterQueryDcl(filterQueryDcl);
        logEventConvert.setFilterQueryDml(filterQueryDml);
        logEventConvert.setFilterQueryDdl(filterQueryDdl);
        logEventConvert.setFilterRows(filterRows);
        logEventConvert.setFilterTableError(filterTableError);
        return logEventConvert;
    }

    public void setEventFilter(EventFilter eventFilter)
    {
        super.setEventFilter(eventFilter);

        // 触发一下filter变更
        if (eventFilter != null && eventFilter instanceof AviaterRegexFilter && binlogParser instanceof LogEventConvert)
        {
            ((LogEventConvert) binlogParser).setNameFilter((AviaterRegexFilter) eventFilter);
        }
    }

    /**
     * 回滚到指定位点
     */
    protected boolean processTableMetaData(EntryPosition entryPosition)
    {
        if (tableMetaTSDB != null)
        {
            if (entryPosition.getTimestamp() == null || entryPosition.getTimestamp() <= 0)
            {
                throw new BinlogParseException("use gtid and TableMetaData TSDB should be config timestamp > 0");
            }

            return tableMetaTSDB.rollback(entryPosition);
        }

        return true;
    }

    public void start() throws BinlogParseException
    {
        if (enableTsdb)
        {
            if (tableMetaTSDB == null)
            {
                synchronized (EventParser.class)
                {
                    try
                    {
                        // 设置当前正在加载的通道，加载spring查找文件时会用到该变量
                        System.setProperty("canal.instance.destination", destination);
                        // 初始化
                        tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml);
                    } finally
                    {
                        System.setProperty("canal.instance.destination", "");
                    }
                }
            }
        }

        super.start();
    }

    public void stop() throws BinlogParseException
    {
        if (enableTsdb)
        {
            tableMetaTSDBFactory.destory(destination);
            tableMetaTSDB = null;
        }

        super.stop();
    }

    public void setEventBlackFilter(EventFilter eventFilter)
    {
        super.setEventBlackFilter(eventFilter);

        // 触发一下filter变更
        if (eventFilter != null && eventFilter instanceof AviaterRegexFilter
                && binlogParser instanceof LogEventConvert)
        {
            ((LogEventConvert) binlogParser).setNameBlackFilter((AviaterRegexFilter) eventFilter);
        }
    }

    protected IMultiStageCoprocessor buildMultiStageCoprocessor()
    {
        MultiStageCoprocessor mysqlMultiStageCoprocessor = new MultiStageCoprocessor(parallelBufferSize,
                parallelThreadSize,
                (LogEventConvert) binlogParser,
                eventTransactionBuffer,
                destination);
        mysqlMultiStageCoprocessor.setEventsPublishBlockingTime(eventsPublishBlockingTime);
        return mysqlMultiStageCoprocessor;
    }

    // ============================ setter / getter =========================

    public void setConnectionCharsetNumber(byte connectionCharsetNumber)
    {
        this.connectionCharsetNumber = connectionCharsetNumber;
    }

    public void setConnectionCharset(Charset connectionCharset)
    {
        this.connectionCharset = connectionCharset;
    }

    public void setConnectionCharset(String connectionCharset)
    {
        this.connectionCharset = Charset.forName(connectionCharset);
    }

    public void setFilterQueryDcl(boolean filterQueryDcl)
    {
        this.filterQueryDcl = filterQueryDcl;
    }

    public void setFilterQueryDml(boolean filterQueryDml)
    {
        this.filterQueryDml = filterQueryDml;
    }

    public void setFilterQueryDdl(boolean filterQueryDdl)
    {
        this.filterQueryDdl = filterQueryDdl;
    }

    public void setFilterRows(boolean filterRows)
    {
        this.filterRows = filterRows;
    }

    public void setFilterTableError(boolean filterTableError)
    {
        this.filterTableError = filterTableError;
    }


    public void setEnableTsdb(boolean enableTsdb)
    {
        this.enableTsdb = enableTsdb;
        if (this.enableTsdb)
        {
            if (tableMetaTSDB == null)
            {
                // 初始化
                tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml);
            }
        }
    }

    public void setTsdbSpringXml(String tsdbSpringXml)
    {
        this.tsdbSpringXml = tsdbSpringXml;
        if (this.enableTsdb)
        {
            if (tableMetaTSDB == null)
            {
                // 初始化
                tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml);
            }
        }
    }

    public void setTableMetaTSDBFactory(TableMetaTSDBFactory tableMetaTSDBFactory)
    {
        this.tableMetaTSDBFactory = tableMetaTSDBFactory;
    }

    public AtomicLong getEventsPublishBlockingTime()
    {
        return this.eventsPublishBlockingTime;
    }

}
